/*
    异步工作器 外界将任务数据添加到输入缓冲区 异步线程对处理缓冲区中的数据进行处理
                若处理缓冲区的数据为0则交换缓冲区


    管理成员：双缓冲区（生产者 消费者）
             互斥锁（保证线程安全）
             条件变量-生产&消费（生产缓冲区没有数据时候 处理完消费缓冲区数据则进入休眠）
             回调函数 针对缓冲区中数据的处理接口 外界接入一个函数 告诉异步工作器这些数据该如何处理
    提供操作：停止异步工作器 添加数据到缓冲区 
            创建线程 交换缓冲区 对消费者缓冲区数据使用回调函数进行处理 处理后再交换
*/
#ifndef __M_LOOPER_H__
#define __M_LOOPER_H__
#include "buffer.hpp"
#include <mutex>
#include <thread>
#include <condition_variable>
#include <functional>
#include <memory>
#include <atomic>

namespace bitlog
{
    using Functor = std::function<void(Buffer &)>; 
    class AsyncLooper
    {
    public:
        using ptr = std::shared_ptr<AsyncLooper>;
        enum class AsyncType{
            ASYNC_SAFE, // 安全状态 当缓冲区满了则阻塞 
            ASYNC_UNSAFE // 无限扩容
        };
        AsyncLooper(const Functor &cb,AsyncType loop_type = AsyncType::ASYNC_SAFE):
            _looper_type(loop_type),
            _stop(false),
            _thread(std::thread(&AsyncLooper::threadEntry,this)),
            _callBack(cb)
        {}
        ~AsyncLooper()
        {
            stop();  
        }
        void stop()
        {
            // 停止异步工作器工作
            _stop = true; // 设置退出标志
            _cond_con.notify_one(); // 唤醒工作线程
            _thread.join();//等待工作线程的退出
        }
        void push(const char* data,size_t len)
        {
            std::unique_lock<std::mutex> lock(_mutex);
            if(_looper_type == AsyncType::ASYNC_SAFE)
            {
                _cond_pro.wait(lock,[&](){ // 条件变量为空值 若缓冲区剩余空间大于数据长度 可添加数据
                return _pro_buf.writeAblesize() >= len;
                 });
            }
            
            // 上面满足后则走下来 可以向缓冲区添加数据
            _pro_buf.push(data,len);
            // 唤醒消费者对缓冲区数据进行处理
            _cond_con.notify_one();
        }
    private:
        //线程入口函数 --对消费缓冲区中的数据进行处理 处理完毕后初始化缓冲区再交换缓冲区
        void threadEntry()
        {
            while(1)
            {
                {
                    // 为互斥锁设置生命周期 缓冲区交换结束后解锁
                    std::unique_lock<std::mutex> lock(_mutex);
                    // 判断退出标志 并且生产缓冲区里面没有数据 才可退出 以防生产缓冲区数据没有完全处理
                    if(_stop && _pro_buf.empty()) break;
                    _cond_con.wait(lock,[&](){
                        return _stop ||  !_pro_buf.empty();
                    }); // 判断生产缓冲区有没有数据          
                    _con_buf.swap(_pro_buf); // 交换缓冲区
                    if(_looper_type == AsyncType::ASYNC_SAFE)
                        _cond_pro.notify_all();// 唤醒生产者
                }
                _callBack(_con_buf); // 交换完后开始处理消费缓冲区
                _con_buf.reset();
            }
        }

    private:
        Functor _callBack; // 回调函数
    private:
        AsyncType _looper_type;
        std::atomic<bool> _stop; // 工作器停止标志
        Buffer _pro_buf; // 生产缓冲区
        Buffer _con_buf; // 消费缓冲区
        std::mutex _mutex;
        std::condition_variable _cond_pro; // 条件变量
        std::condition_variable _cond_con;
        std::thread _thread; // 异步工作器对应的工作线程
    };
}

#endif